/**
 * Software License, Version 1.0
 * 
 * Copyright 2008 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgment:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgment may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or Granules, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called Granules, 
 * nor may Indiana University or Community Grids Lab or Granules appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 * GENERATED USING SOFTWARE.
 */
package cgl.granules.samples.grep;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Properties;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import cgl.granules.dataset.Dataset;
import cgl.granules.dataset.DatasetCollection;
import cgl.granules.dataset.DatasetException;
import cgl.granules.dataset.DatasetFactory;
import cgl.granules.dataset.FileAccess;
import cgl.granules.exception.MarshallingException;
import cgl.granules.operation.MapReduceBase;
import cgl.granules.operation.MapReduceException;
import cgl.granules.results.ResultPayload;
import cgl.granules.results.Results;

/**
 * An example demonstrating the <code>Map</code> role within a Grep Count
 * operation.
 * 
 * @author Shrideep Pallickara
 * 
 */
public class GrepCountMap extends MapReduceBase {

    private Hashtable<String, Integer> counts =
        new Hashtable<String, Integer>();
    private ResultPayload resultPayload;
    private Pattern pattern;
    private int grepCount = 0;
    // private String stringToGrepFor = "belittle";
    private String stringToGrepFor = "twinkle";
    String workerType = "Unknown ";


    public GrepCountMap() {
        setDomain("Generic");
        setVersionInformation(200);
        setOperationIdentifier("GrepCount-MapFunction-GranulesSample");
        resultPayload = ResultPayload.getInstance();
    }


    public void execute() {
        Properties props = getProcessingDirectives();

        String property = "Worker Number";
        if (props.containsKey(property)) {
            workerType = property + props.getProperty(property);
        }

       
        if (!hasDatasetCollection()) {
            System.out.println("No dataset has been initialized. Returning ...");
            return;
        }

        String datasetIdentifier = "FileDataset-ID-GrepCount";
        DatasetCollection datasetCollection = getDatasetCollection();
        FileAccess access = null;
        try {
            Dataset dataset = datasetCollection.getDataset(datasetIdentifier);

            if (dataset.getDatasetType() == Dataset.FILES) {
                access = (FileAccess) dataset;
            } else {
                System.out.println("Incorrect datatset: "
                    + dataset.getDatasetType() + " initialized. Returning ...");
                return;
            }

            if (access == null) {
                System.out.println("Unable to initialize the dataset");
                return;
            }

            Enumeration filenames = access.getFilenames();
            while (filenames.hasMoreElements()) {
                String filename = (String) filenames.nextElement();
                parseAndReadFromFile(filename);
            }

            ResultPayload resultPayload = ResultPayload.getInstance();
            byte[] resultPayloadBytes = resultPayload.getBytes(counts);

            Results results = createResults(true, true);

            results.setResultPayload(resultPayloadBytes);
            writeResults(results);
            setTeminationConditionReached();

        } catch (DatasetException e) {
            e.printStackTrace();
        } catch (IOException e) {
            System.out.println("Problems converting bytes to longs");
            e.printStackTrace();
        } catch (MarshallingException e) {
            System.out.println("Problems Marshalling: " + e.getMessage());
            e.printStackTrace();
        } catch (MapReduceException e) {
            System.out.println("Problems writing Results: " + e.getMessage());
            e.printStackTrace();
        }
        // System.out.println("Finished :" + workerType + ", which did ("
        // + workUnits + ") units of work on the data.");

    }


    private void parseAndReadFromFile(String filename) throws DatasetException,
        IOException {
        DatasetFactory datasetFactory = DatasetFactory.getInstance();
        FileAccess dataset =
            datasetFactory.createFileDataset("Test", "Grep Count example");

        String line = "";
        BufferedReader in = new BufferedReader(new FileReader(filename), 65536);
        pattern = Pattern.compile(stringToGrepFor);
        while ((line = in.readLine()) != null) {
            parse(line);
        }

        counts.put(stringToGrepFor, grepCount);
        in.close();
    }


    private void parse(String text) {
        Matcher matcher = pattern.matcher(text);
        while (matcher.find()) {
            grepCount++;
        }

    }

}
